package com.xy.job.admin.service

import com.xy.job.admin.api.request.RegisteredNodeQueryParam
import com.xy.job.admin.common.ADMIN_MASTER_LOCK
import com.xy.job.admin.common.ADMIN_NODE_NAME
import com.xy.job.admin.common.NODE_EXPIRED_SECONDS
import com.xy.job.admin.domain.NodeRegistry
import com.xy.job.admin.enums.NodeType
import com.xy.job.admin.repository.JobInstanceRepository
import com.xy.job.admin.repository.JobLockRepository
import com.xy.job.admin.repository.NodeRegistryRepository
import org.slf4j.LoggerFactory
import org.springframework.data.domain.*
import org.springframework.stereotype.Service
import org.springframework.transaction.annotation.Transactional
import java.time.LocalDateTime

@Service
class NodeRegistryService(
    private var lockRepository: JobLockRepository,
    private var jobInstanceRepository: JobInstanceRepository,
    private var nodeRegistryRepository: NodeRegistryRepository
) {

    private val log = LoggerFactory.getLogger(this.javaClass)

    fun isAdminMasterNode(nodeAddress: String): Boolean {
        val adminMasterNodeRegistry = findAdminMasterNodeRegistry()
        return when {
            adminMasterNodeRegistry == null -> false
            else -> adminMasterNodeRegistry.nodeAddress == nodeAddress
        }
    }

    @Transactional
    fun cleanDeadNodeRegistry() {
        val lastHeartbeatTime = LocalDateTime.now().minusSeconds(NODE_EXPIRED_SECONDS)
        val deadNodeRegistryIterable = nodeRegistryRepository.findByLastHeartbeatTimeLessThan(lastHeartbeatTime)
        val nodeAddressList = deadNodeRegistryIterable.map { it.nodeAddress!! }
        if (nodeAddressList.isNotEmpty()) {
            nodeRegistryRepository.deleteAll(deadNodeRegistryIterable)
            lockRepository.resetLockByOwner(nodeAddressList)
            jobInstanceRepository.markFailedByAddress(
                nodeAddressList = nodeAddressList,
                completeTime = LocalDateTime.now(),
                errorMessage = "executor was offline",
            )
        }
    }

    fun refreshNodeRegistry(nodeAddress: String) {
        val nodeRegistryIterable = nodeRegistryRepository.findByNodeAddress(nodeAddress = nodeAddress)
        nodeRegistryIterable.forEach { it.lastHeartbeatTime = LocalDateTime.now() }
        nodeRegistryRepository.saveAll(nodeRegistryIterable)
    }

    fun registerAdminScheduleNode(host: String, port: Int) {
        val nodeAddress = "$host:$port"
        val nodeRegistry = nodeRegistryRepository.findByNodeAddress(nodeAddress = nodeAddress)
            .firstOrNull { it.nodeType == NodeType.ADMIN_SCHEDULER }
        when (nodeRegistry == null) {
            true -> {
                val newNodeRegistry = NodeRegistry(
                    nodeHost = host,
                    nodePort = port,
                    nodeType = NodeType.ADMIN_SCHEDULER,
                    nodeName = ADMIN_NODE_NAME,
                    nodeAddress = nodeAddress,
                    registerTime = LocalDateTime.now(),
                    lastHeartbeatTime = LocalDateTime.now(),
                )
                nodeRegistryRepository.save(newNodeRegistry)
            }

            false -> {}
        }
    }

    @Transactional(timeout = 2)
    fun registerAdminMasterNode(host: String, port: Int) {
        if (existAliveAdminMasterNodeRegistry()) {
            return
        }

        lockRepository.lock(ADMIN_MASTER_LOCK)
        val adminMasterNodeRegistry = findAdminMasterNodeRegistry()
        when {
            // register self as master node
            adminMasterNodeRegistry == null -> registerSelfAsAdminNode(host, port)
            else -> {
                val adminNodeExpiredTime = adminMasterNodeRegistry.lastHeartbeatTime!!.plusSeconds(NODE_EXPIRED_SECONDS)
                val oldAdminMasterNodeIsAlive = adminNodeExpiredTime > LocalDateTime.now()
                when (oldAdminMasterNodeIsAlive) {
                    // if admin master node still alive, then do nothing
                    true -> Unit
                    // if admin master node is dead, then try to replace it
                    false -> {
                        nodeRegistryRepository.deleteById(adminMasterNodeRegistry.id!!)
                        registerSelfAsAdminNode(host, port)
                        log.info("replace old master node success")
                    }
                }
            }
        }
    }

    private fun registerSelfAsAdminNode(host: String, port: Int) {
        val newAdminMasterNodeRegistry = NodeRegistry(
            nodeHost = host,
            nodePort = port,
            nodeType = NodeType.ADMIN_MASTER,
            nodeName = ADMIN_NODE_NAME,
            nodeAddress = "$host:$port",
            registerTime = LocalDateTime.now(),
            lastHeartbeatTime = LocalDateTime.now(),
        )
        nodeRegistryRepository.save(newAdminMasterNodeRegistry)
        log.info("register self as master node success")
    }

    private fun findAdminMasterNodeRegistry(): NodeRegistry? {
        return nodeRegistryRepository.findByNodeType(nodeType = NodeType.ADMIN_MASTER).firstOrNull()
    }

    private fun existAliveAdminMasterNodeRegistry(): Boolean {
        return nodeRegistryRepository.findByNodeTypeAndLastHeartbeatTimeGreaterThan(
            nodeType = NodeType.ADMIN_MASTER,
            lastHeartbeatTime = LocalDateTime.now().minusSeconds(NODE_EXPIRED_SECONDS),
        ).any()
    }

    fun listRegistry(param: RegisteredNodeQueryParam): Page<NodeRegistry> {
        val pageable: Pageable = PageRequest.of(
            param.pageNo,
            param.pageSize,
            Sort.by(NodeRegistry::nodeType.name, NodeRegistry::registerTime.name).descending()
        )
        val nodeRegistry = NodeRegistry().apply {
            this.nodeName = param.nodeName
            this.nodeType = param.nodeType
        }
        val matcher = ExampleMatcher.matchingAll()
            .withMatcher(NodeRegistry::nodeName.name, ExampleMatcher.GenericPropertyMatchers.contains())
            .withMatcher(NodeRegistry::nodeType.name, ExampleMatcher.GenericPropertyMatchers.exact())
        val example = Example.of(nodeRegistry, matcher)
        return nodeRegistryRepository.findAll(example, pageable)
    }

    fun removeNodeRegistry(nodeAddress: String) {
        val nodeRegistries = nodeRegistryRepository.findByNodeAddress(nodeAddress)
        nodeRegistryRepository.deleteAll(nodeRegistries)
    }
}